package com.study.rocketmq.service.impl;

import com.study.rocketmq.message.MQMessage;
import com.study.rocketmq.service.IRocketMQService;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;


@Service
public class RocketMQServiceImpl implements IRocketMQService {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQServiceImpl.class);

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public boolean send(MQMessage message) {
        if (logger.isDebugEnabled()) logger.info("send() message={}", message);

        SendResult sendResult = rocketMQTemplate.syncSend(message.getTopic() + ":" + message.getTags(),
                MessageBuilder.withPayload(message.getContent()).build());
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            return true;
        }
        if (logger.isDebugEnabled()) logger.info("send() sendResult={}", sendResult);
        return false;
    }

    @Override
    public void asyncSend(MQMessage message) {
        if (logger.isDebugEnabled()) logger.info("asyncSend(): message= {}", message);

        rocketMQTemplate.asyncSend(message.getTopic() + ":" + message.getTags(), message.getContent(),
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        /** 成功不做日志记录或处理 */
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        if (logger.isDebugEnabled()) logger.info("asyncSend(): fail message={}", message);
                    }
                });

    }

    @Override
    public void syncSendOrderly(MQMessage message) {
        if (logger.isDebugEnabled()) logger.info("syncSendOrderly(): message= {}", message);
        rocketMQTemplate.sendOneWay(message.getTopic() + ":" + message.getTags(), message.getContent());

    }
}
